﻿using System;
using System.Collections.Generic;
using MySql.Data.MySqlClient;
using System.Text;
using System.Data;
using Microsoft.WindowsAzure.ServiceRuntime;
using System.IO;
using System.Diagnostics;
using System.Threading;
using System.Net;

namespace HWCWorker_Worker
{

    internal class BinLogInfo
    {
        public int LogId { get; set; }
        public long Position { get; set; }
        public string HostInstanceId { get; set; }
        public RoleInstanceEndpoint EndPoint { get; set; }
    }

    class BinLogInfoSorter : IComparer<BinLogInfo>
    {
        public int Compare(BinLogInfo left, BinLogInfo right)
        {
            if (left == null || right == null)
            {
                throw new ArgumentNullException("ChunkSorter");
            }
            int compare = left.LogId.CompareTo(right.LogId);
            if (compare != 0)
            {
                return compare;
            }

            // break the tie with log position
            compare = left.Position.CompareTo(right.Position);
            if (compare != 0)
            {
                return compare;
            }

            // break the tie with role instance id
            return left.HostInstanceId.CompareTo(right.HostInstanceId);
        }
    }

    internal class MySQLClient
    {
        private string _endpointName;
        //mysql command to get the slave status
        private string _checkSlaveCmd = "show slave status";
        //mysql command to get the server id
        private string _getServerIdCmd = "select @@server_id";
        //mysql command to to get the master status
        private string _showMasterCmd = "show master status";
        //mysql command to start the slave
        private string _startSlaveCmd = "start slave";
        //mysql command to stop the slave
        private string _stopSlaveCmd = "stop slave";
        //mysql command to reset the slave
        private string _resetSlaveCmd = "RESET SLAVE";
        //mysql command to reset master
        private string _resetMasterCmd = "RESET MASTER";
        //mysql command to make a slave replicate from the master
        private string _slaveFmt = "CHANGE MASTER TO MASTER_HOST='{0}', MASTER_PORT={1}, MASTER_USER='{2}', MASTER_PASSWORD='{3}'";
        //mysql command to make a slave replicate from the master
        private string _changeSlaveFmt = "CHANGE MASTER TO MASTER_HOST='{0}', MASTER_PORT={1}, MASTER_USER='{2}', MASTER_PASSWORD='{3}', MASTER_LOG_FILE='{4}', MASTER_LOG_POS={5}";
        //mysql command to grant replication to slaves from master
        private string _grantRepl = "GRANT REPLICATION SLAVE ON *.* TO 'root' ";
        //mysql command to grant replication on slaves from master
        private string _grantReplTest = "GRANT REPLICATION SLAVE ON *.* TO '{0}'@'%' IDENTIFIED BY '{1}'";
        //mysql command to grant permissions to a user
        private string _grantPerm = "GRANT ALL PRIVILEGES ON *.* TO '{0}'@'%' IDENTIFIED BY '{1}' WITH GRANT OPTION";
        //mysql command to grant permissions to a user
        private string _grantRoot = "GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' WITH GRANT OPTION";
        //mysql command to grant permissions to a user
        private string _grantRootLocal = "GRANT ALL PRIVILEGES ON *.* TO 'root'@'localhost' WITH GRANT OPTION";
        //mysql command to grant permissions to a user
        private string _grantLocal = "GRANT ALL PRIVILEGES ON *.* TO '{0}'@'localhost' IDENTIFIED BY '{1}' WITH GRANT OPTION";
        //mysql command to set read only on a server
        private string _setReadOnly = "SET GLOBAL read_only = ON";
        //mysql command to switch off read only on a server
        private string _removeReadOnly = "SET GLOBAL read_only = OFF";
        //mysql command to lock the tables with read-only on a server
        private string _setReadLock = "FLUSH TABLES WITH READ LOCK";
        //mysql command to unlock the tables
        private string _unlockTables = "UNLOCK TABLES";

        //TODO:use master_password
        private static object _syncObj = new object();
        private static MySqlConnection _connection = null;
        private LogClient _logger;

        public MySQLClient(string endpointName)
        {
            _endpointName = endpointName;
            _logger = new LogClient(RoleEnvironment.CurrentRoleInstance.Id);
        }

        /// <summary>
        /// Checks if the mysql instance is running 
        /// </summary>
        /// <param name="endpointName"></param>
        /// <returns></returns>
        public static bool IsRunning(string endpointName)
        {
            bool isActive = false;
            MySqlConnection rootConn = GetConnection(endpointName);
            if (rootConn == null)
            {
                return false;
            }
            lock (_syncObj)
            {
                try
                {
                    //Pings the mysql server to check if it is active
                    isActive = rootConn.Ping();
                }
                catch (Exception)
                {
                    isActive = false;
                }

                if (!isActive)
                {
                    _connection = null;
                }
            }
            return isActive;
        }

        /// <summary>
        /// Checks if the mysql instance is a master
        /// </summary>
        /// <returns></returns>
        public bool IsMaster()
        {
            if (!IsRunning(_endpointName))
            {
                return false;
            }
            MySqlConnection rootConn = GetConnection(_endpointName);
            MySqlCommand cmd = new MySqlCommand(_checkSlaveCmd, rootConn);
            MySqlDataReader reader = null;
            try
            {
                reader = cmd.ExecuteReader();
                if (reader.HasRows)
                {
                    string ioRunning = reader["Slave_IO_Running"].ToString();
                    string sqlRunning = reader["Slave_SQL_Running"].ToString();
                    if (ioRunning == "No" && sqlRunning == "No")
                    {
                        return true;
                    }
                    else
                    {
                        // this is a slave node 
                        return false;
                    }
                }
            }
            catch (MySqlException)
            {
                return false;
            }
            return true;
        }

        /// <summary>
        /// Gets the mysql server id for a mysql instance
        /// </summary>
        /// <returns></returns>
        public int GetMySqlServerId()
        {
            if (!IsRunning(_endpointName))
            {
                // MySql is not running yet
                // TODO: get server id from my.ini
                return -1;
            }
            MySqlConnection rootConn = GetConnection(_endpointName);
            MySqlCommand cmd = new MySqlCommand(_getServerIdCmd, rootConn);
            MySqlDataReader reader = null;
            try
            {
                reader = cmd.ExecuteReader();
                if (reader.Read())
                {
                    // this is MySQL's server id
                    return reader.GetInt16(0);
                }
            }
            catch (MySqlException ex)
            {
                LogError("Error in GetMySqlServerId: {0}", ex.Message + ex.StackTrace);
            }
            return -1;
        }

        /// <summary>
        /// Gets the bin log id and position for a mysql instance
        /// </summary>
        /// <returns></returns>
        public BinLogInfo GetBinLogInfo()
        {
            if (!IsRunning(_endpointName))
            {
                return GetBinLogInfoFromFile();
            }
            MySqlConnection rootConn = GetConnection(_endpointName);
            MySqlCommand cmd = new MySqlCommand(_showMasterCmd, rootConn);
            MySqlDataReader reader = null;
            BinLogInfo logInfo = null;
            try
            {
                reader = cmd.ExecuteReader();
                if (reader.Read())
                {
                    // e.g. logFile (mysql-bin.000001, mysql-bin.000002)
                    string logFile = reader.GetString(0);
                    int pos = logFile.LastIndexOf('.');
                    if (pos == -1)
                    {
                        return null;
                    }
                    string logIdStr = logFile.Substring(pos + 1);
                    int logId = 0;
                    if (!int.TryParse(logIdStr, out logId))
                    {
                        return null;
                    }
                    logInfo = new BinLogInfo();
                    logInfo.LogId = logId;
                    logInfo.Position = reader.GetInt64(1);
                }
            }
            catch (MySqlException ex)
            {
                LogError("Error in GetBinLogInfo: {0}", ex.Message + ex.StackTrace);
            }
            return logInfo;
        }

        /// <summary>
        /// Changes a mysql instance as slave to a master
        /// </summary>
        /// <param name="masterHost"></param>
        /// <param name="masterPort"></param>
        /// <returns></returns>
        public bool ChangeToSlave(string masterHost, int masterPort, string userName, string password)
        {
            bool status = false;
            try
            {
                if (!IsRunning(_endpointName))
                {
                    return false;
                }
                string slaveCmd = string.Format(_slaveFmt, masterHost, masterPort, userName, password);
                MySqlConnection rootConn = GetConnection(_endpointName);
                //stop slave
                MySqlCommand cmd = new MySqlCommand(_stopSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //reset slave
                cmd = new MySqlCommand(_resetSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //Change master to command
                cmd = new MySqlCommand(slaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //start slave
                cmd = new MySqlCommand(_startSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //grant permissions to remote root user
                cmd = new MySqlCommand(_grantRoot, rootConn);
                cmd.ExecuteNonQuery();
                //grant permissions to local root user
                cmd = new MySqlCommand(_grantRootLocal, rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to mysql user
                cmd = new MySqlCommand(string.Format(_grantPerm, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to local mysql user
                cmd = new MySqlCommand(string.Format(_grantLocal, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //set the global variable read_only to true
                cmd = new MySqlCommand(_setReadOnly, rootConn);
                cmd.ExecuteNonQuery();

                status = true;
            }
            catch (MySqlException ex)
            {
                LogError("Error in ChangeToSlave: {0}", ex.Message + ex.StackTrace);
                return false;
            }
            return status;
        }

        /// <summary>
        /// changes a mysql instance to master
        /// </summary>
        /// <returns></returns>
        public bool ChangeToMaster(string userName, string password)
        {
            bool status = false;
            try
            {
                if (!IsRunning(_endpointName))
                {
                    return false;
                }
                MySqlConnection rootConn = GetConnection(_endpointName);
                //stop slave
                MySqlCommand cmd = new MySqlCommand(_stopSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //reset master
                cmd = new MySqlCommand(_resetMasterCmd, rootConn);
                cmd.ExecuteNonQuery();
                //grant replication privilege to slaves
                cmd = new MySqlCommand(_grantRepl, rootConn);
                cmd.ExecuteNonQuery();
                //grant permissioon to remote root user
                cmd = new MySqlCommand(_grantRoot, rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to mysql user
                cmd = new MySqlCommand(string.Format(_grantPerm, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to local root user
                cmd = new MySqlCommand(_grantRootLocal, rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to local mysql user
                cmd = new MySqlCommand(string.Format(_grantLocal, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //grant replication privilege to mysql user on slaves
                cmd = new MySqlCommand(string.Format(_grantReplTest, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                string conn = rootConn.ConnectionString;
                status = true;
            }
            catch (MySqlException ex)
            {
                LogError("Error in ChangeToMaster: {0}", ex.Message + ex.StackTrace);
                return false;
            }
            return status;
        }

        /// <summary>
        /// This method copies the mysql binaries from solution to role's local storage
        /// and starts the mysql instance suing mysqld
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        public bool Start(int id)
        {
            try
            {
                string baseDir = RoleEnvironment.GetLocalResource("MySQL").RootPath.Replace('\\', '/');
                string dataDir = RoleEnvironment.GetLocalResource("MySQLDatastore").RootPath.Replace('\\', '/');
                string blobDir = RoleEnvironment.GetLocalResource("BlobBackup").RootPath.Replace('\\', '/');
                LogError("MySql Base directory: {0}", baseDir);

                LogError("MySql Data directory: {0}", dataDir);

                string command = Path.Combine(baseDir, @"bin\mysqld.exe");
                SetupBlobBackup(blobDir);
                if (!File.Exists(command))
                {
                    SetupAzureMySql(baseDir);
                    SetupAzureMySqlDataDir(dataDir);
                }
                MySqlConnection rootConn = GetConnection(_endpointName);
                if (IsRunning(_endpointName))
                {
                    int currentId = GetMySqlServerId();
                    if (currentId == id)
                    {
                        return true;
                    }
                    //TODO:make sure shutdown existing SQL
                    command = Path.Combine(baseDir, @"bin\mysqladmin.exe");
                    Process.Start(command, "-u root shutdown");
                }

                RoleInstanceEndpoint instance;
                instance = RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["MySQL"];
                string port = instance.IPEndpoint.Port.ToString();
                string iniFile = Path.Combine(baseDir, "my.ini");
                File.Copy("my.ini", iniFile, true);
                //update teh my.ini file with mysql server details
                UpdateMyIni(iniFile, baseDir, dataDir, port, id.ToString());
                ProcessStartInfo startInfo = new ProcessStartInfo(command);
                startInfo.RedirectStandardOutput = true;
                startInfo.WorkingDirectory = baseDir;
                startInfo.UseShellExecute = false;
                startInfo.CreateNoWindow = true;
                Process driver = new Process();
                driver.StartInfo = startInfo;
                driver.Start();
                
                StreamReader sr = driver.StandardOutput;
                string output = sr.ReadToEnd();
                while (!IsRunning(_endpointName))
                {
                    Thread.Sleep(TimeSpan.FromSeconds(10));
                }
            }
            catch (Exception ex)
            {
                LogError("Error in MySqlAccess start(): {0}", ex.Message + ex.StackTrace);
                return false;
            }
            return true;
        }

        /// <summary>
        /// Invoke native methods to update the my.ini file
        /// </summary>
        /// <param name="iniFile"></param>
        /// <param name="baseDir"></param>
        /// <param name="dataDir"></param>
        /// <param name="port"></param>
        /// <param name="id"></param>
        private void UpdateMyIni(string iniFile, string baseDir, string dataDir, string port, string id)
        {
            NativeMethods.WritePrivateProfileString("mysqld", "basedir", baseDir, iniFile);
            NativeMethods.WritePrivateProfileString("mysqld", "datadir", dataDir, iniFile);
            NativeMethods.WritePrivateProfileString("mysqld", "innodb_data_home_dir", dataDir, iniFile);
            NativeMethods.WritePrivateProfileString("mysqld", "port", port, iniFile);
            NativeMethods.WritePrivateProfileString("mysqld", "server-id", id, iniFile);
            NativeMethods.WritePrivateProfileString("client", "port", port, iniFile);
        }

        // example input: mysql-bin.000001
        //                deployment(808).webPortalV2.webPortal_WebRole.0
        public static int GetIdFromName(string inputName)
        {
            int id = -1;
            int pos = inputName.LastIndexOf('.');
            if (pos == -1)
            {
                return id;
            }
            string idStr = inputName.Substring(pos + 1);
            if (!int.TryParse(idStr, out id))
            {
                return -1;
            }
            return id;
        }

        /// <summary>
        /// Copies mysql binaries to root folder
        /// </summary>
        /// <param name="rootDir"></param>
        /// <returns></returns>
        private bool SetupAzureMySql(string rootDir)
        {
            if (!Directory.Exists("mysql"))
            {
                return false;
            }
            CopyDir("mysql", rootDir);
            return true;
        }

        /// <summary>
        /// copies mysql data folder 
        /// </summary>
        /// <param name="dataDir"></param>
        /// <returns></returns>
        private bool SetupAzureMySqlDataDir(string dataDir)
        {
            string srcData = @"mysql\data";
            if (!Directory.Exists(srcData))
            {
                return false;
            }
            CopyDir(srcData, dataDir);
            return true;
        }


        /// <summary>
        /// copies blob backup
        /// </summary>
        /// <param name="dataDir"></param>
        /// <returns></returns>
        private bool SetupBlobBackup(string blobDir)
        {
            if (!Directory.Exists("BlobStorageBackup"))
            {
                return false;
            }
            CopyDir("BlobStorageBackup", blobDir);
            return true;
        }

        /// <summary>
        /// Copies the files and folders recursively from source to destination
        /// </summary>
        /// <param name="src"></param>
        /// <param name="dst"></param>
        /// <returns></returns>
        private static bool CopyDir(string src, string dst)
        {
            try
            {
                if (!Directory.Exists(dst))
                {
                    Directory.CreateDirectory(dst);
                }
                string[] subDirs = Directory.GetDirectories(src);
                foreach (string dir in subDirs)
                {
                    int pos = dir.LastIndexOf('\\');
                    string target = Path.Combine(dst, dir.Substring(pos + 1));
                    CopyDir(dir, target);
                }
                string[] files = Directory.GetFiles(src);
                foreach (string file in files)
                {
                    string targetFile = Path.Combine(dst, Path.GetFileName(file));
                    File.Copy(file, targetFile);
                }
            }
            catch (Exception ex)
            {
                return false;
            }
            return true;
        }

        /// <summary>
        /// Get the mysqlconenction for an endpoint
        /// </summary>
        /// <param name="epName"></param>
        /// <returns></returns>
        private static MySqlConnection GetConnection(string epName)
        {
            if (_connection != null)
            {
                return _connection;
            }
            lock (_syncObj)
            {
                try
                {
                    RoleInstanceEndpoint instance;
                    instance = RoleEnvironment.CurrentRoleInstance.InstanceEndpoints[epName];
                    // only local connection to MySql is allowed, 
                    StringBuilder sb = new StringBuilder("host=localhost;uid=root;");
                    sb.AppendFormat("protocol=tcp;port={0}", instance.IPEndpoint.Port);
                    // Note:the following code tries to access via network
                    //sb.AppendFormat("server={0};user id=root", instance.IPEndPoint.Address);
                    _connection = new MySqlConnection(sb.ToString());
                    _connection.Open();
                }
                catch (Exception ex)
                {
                    LogClient logger = new LogClient(RoleEnvironment.CurrentRoleInstance.Id);
                    logger.Log("GetConnection:Error: {0}", ex.Message + ex.StackTrace);
                    _connection = null;
                }
            }
            return _connection;
        }

        /// <summary>
        /// Gets the binloginfo from bin log file
        /// </summary>
        /// <returns></returns>
        private BinLogInfo GetBinLogInfoFromFile()
        {
            try
            {
                BinLogInfo logInfo = new BinLogInfo()
                {
                    LogId = 0,
                    Position = 0
                };
                string sqlDataRoot = RoleEnvironment.GetLocalResource("MySQLDatastore").RootPath;
                string indexFile = Path.Combine(sqlDataRoot, "mysql-bin.index");
                // first check if MySql is located at local storage
                if (!File.Exists(indexFile))
                {
                    indexFile = Path.Combine(@"mysql\data\", "mysql-bin.index");
                    // then check deployment disk for MySql's location
                    if (!File.Exists(indexFile))
                    {
                        return logInfo;
                    }
                }

                string binLogFile = null;
                int retry = 0;
                while (binLogFile == null)
                {
                    // the index file may not be accessible for one try
                    binLogFile = GetBinLogFile(indexFile);
                    Thread.Sleep(50);
                    if (++retry > 3)
                    {
                        return logInfo;
                    }
                }
                int logId = GetIdFromName(binLogFile);
                if (logId == -1)
                {
                    throw new Exception("MySql configuration error!");
                }
                // get full path of bin-log file
                binLogFile = Path.Combine(sqlDataRoot, binLogFile);
                //binLogFile = Path.Combine(@"mysql\data\", binLogFile);
                if (!File.Exists(binLogFile))
                {
                    LogError("MySql configuration error: missing--> {0}", binLogFile);
                    throw new Exception("MySql configuration error: missing-->" + binLogFile);
                }
                else
                {
                    LogError("GetBinLogInfoFromFile:Binlog file does not exist");
                }
                FileInfo logFileInfo = new FileInfo(binLogFile);
                logInfo.LogId = logId;
                logInfo.Position = logFileInfo.Length;
                return logInfo;
            }
            catch (Exception ex)
            {
                LogError("Error in GetBinLoginfoFromFile: {0}", ex.Message + ex.StackTrace);
                return null;
            }
        }

        /// <summary>
        /// Get the bin log file 
        /// </summary>
        /// <param name="indexFile"></param>
        /// <returns></returns>
        private string GetBinLogFile(string indexFile)
        {
            string binLogFile = null;
            try
            {
                using (StreamReader sr = File.OpenText(indexFile))
                {
                    string line;
                    while ((line = sr.ReadLine()) != null)
                    {
                        //scan until the last line
                        binLogFile = line;
                    }
                    sr.Close();
                    sr.Dispose();
                }
            }
            catch (Exception ex)
            {
                LogError("Error in GetBinLogFile: {0}", ex.Message + ex.StackTrace);
                //RoleManager.WriteToLog("Error", ex.Message);                
            }
            return binLogFile;
        }

        #region Added for failover (Infosys)

        /// <summary>
        /// Get a mysql connection
        /// </summary>
        /// <param name="address"></param>
        /// <param name="port"></param>
        /// <returns></returns>
        public static MySqlConnection GetMySqlConnection(string address, int port)
        {
            LogClient logger = new LogClient(RoleEnvironment.CurrentRoleInstance.Id);
            StringBuilder sb = new StringBuilder();
            sb.AppendFormat("server={0};user id=root;", address);
            sb.AppendFormat("protocol=tcp;port={0};pooling=false;", port);
            logger.Log("GetMySqlConnection: ConnString: {0}", sb.ToString());
            MySqlConnection _MySqlconnection = new MySqlConnection(sb.ToString());
            try
            {
                _MySqlconnection.Open();
            }
            catch (Exception ex)
            {
                logger.Log("GetMySqlConnection:Error: {0}", ex.Message + ex.StackTrace);
                logger.Log("GetMySqlConnection:InnerException: {0}", ex.InnerException);
                _MySqlconnection = null;
            }

            return _MySqlconnection;
        }

        /// <summary>
        /// Make slaves replicate from new master
        /// </summary>
        /// <param name="masterHost"></param>
        /// <param name="masterPort"></param>
        /// <param name="slaveHost"></param>
        /// <param name="slavePort"></param>
        /// <returns></returns>
        public bool ChangeSlavesMaster(string masterHost, int masterPort, string slaveHost, int slavePort, string userName, string password)
        {
            LogError("Inside change slaves master method");
            MySqlConnection masterConn = GetMySqlConnection(masterHost, masterPort);
            string masterLogFile = string.Empty;
            int masterLogPos = 0;
            string slaveCmd = string.Empty;
            if (!IsMySqlRunning(masterHost, masterPort))
            {
                return false;
            }

            MySqlCommand command = new MySqlCommand(_showMasterCmd, masterConn);
            using (MySqlDataReader reader = command.ExecuteReader())
            {
                if (reader.HasRows)
                {
                    while (reader.Read())
                    {
                        masterLogFile = reader["File"].ToString().ToLower();
                        masterLogPos = Convert.ToInt32(reader["Position"]);
                    }
                }
            }
            LogError("ChangeSlavesMaster: MasterLogFile:MasterLogPos: {0}:{1}", masterLogFile, masterLogPos);

            if (!string.IsNullOrEmpty(masterLogFile) && masterLogPos != 0)
            {
                slaveCmd = string.Format(_changeSlaveFmt, masterHost, masterPort, userName, password, masterLogFile, masterLogPos);
                LogError("ChangeSlavesMaster: SlaveComamnd: {0}", slaveCmd);
            }
            else
            {
                slaveCmd = string.Format(_slaveFmt, masterHost, masterPort, userName, password);
                LogError("ChangeSlavesMaster: SlaveComamnd: {0}", slaveCmd);
            }

            MySqlConnection rootConn = GetMySqlConnection(slaveHost, slavePort);//GetConnection(_endpointName);
            bool status = false;
            try
            {
                if (!IsMySqlRunning(slaveHost, slavePort))
                {
                    return false;
                }
                //stop slave
                LogError("ChangeSlavesMaster:StopSlave");
                MySqlCommand cmd = new MySqlCommand(_stopSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //reset slave
                LogError("ChangeSlavesMaster:Reset slave");
                cmd = new MySqlCommand(_resetSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //change master to
                LogError("ChangeSlavesMaster:Change master");
                cmd = new MySqlCommand(slaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //start slave
                LogError("ChangeSlavesMaster:StartSlave");
                cmd = new MySqlCommand(_startSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                //grant permissions to remote root user
                LogError("ChangeSlavesMaster:Grantroot");
                cmd = new MySqlCommand(_grantRoot, rootConn);
                cmd.ExecuteNonQuery();
                //grant permissions to local root user
                LogError("ChangeSlavesMaster:GrantRootLocal");
                cmd = new MySqlCommand(_grantRootLocal, rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to mysql user
                LogError("ChangeSlavesMaster:GrantPerm");
                cmd = new MySqlCommand(string.Format(_grantPerm, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //grant privilege to local mysql user
                LogError("ChangeSlavesMaster:GrantLocal");
                cmd = new MySqlCommand(string.Format(_grantLocal, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                //set the global variable read_only to true
                LogError("ChangeSlavesMaster:SetReadOnly");
                cmd = new MySqlCommand(_setReadOnly, rootConn);
                cmd.ExecuteNonQuery();
                status = true;
            }
            catch (MySqlException ex)
            {
                LogError("Error in ChangeSlavesMaster: {0}", ex.Message + ex.StackTrace);
            }
            masterConn.Close();
            rootConn.Close();
            return status;
        }

        /// <summary>
        /// Promote slave as master
        /// </summary>
        /// <param name="host"></param>
        /// <param name="port"></param>
        /// <returns></returns>
        public bool ChangeSlaveToMaster(string host, int port, string userName, string password)
        {
            LogError("Inside ChangeSlaveToMaster");
            MySqlConnection rootConn = GetMySqlConnection(host, port);//GetConnection(_endpointName);
            MySqlCommand cmd = new MySqlCommand(_stopSlaveCmd, rootConn);
            bool status = false;
            try
            {
                if (!IsMySqlRunning(host, port))
                {
                    return false;
                }
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(_resetMasterCmd, rootConn);
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(_resetSlaveCmd, rootConn);
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(_grantRepl, rootConn);
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(_removeReadOnly, rootConn);
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(_unlockTables, rootConn);
                cmd.ExecuteNonQuery();
                cmd = new MySqlCommand(string.Format(_grantReplTest, userName, password), rootConn);
                cmd.ExecuteNonQuery();
                string conn = rootConn.ConnectionString;
                status = true;
            }
            catch (MySqlException ex)
            {
                LogError("Error in ChangeSlaveToMaster: {0}", ex.Message + ex.StackTrace);
            }
            rootConn.Close();
            return status;
        }

        /// <summary>
        /// Get the bin log info for a slave instance
        /// </summary>
        /// <param name="host"></param>
        /// <param name="port"></param>
        /// <param name="instanceId"></param>
        /// <param name="ep"></param>
        /// <returns></returns>
        public BinLogInfo GetBinLogInfoForSlave(string host, int port, string instanceId, RoleInstanceEndpoint ep)
        {
            MySqlConnection rootConn = GetMySqlConnection(host, port);
            //execute show slave status command
            MySqlCommand cmd = new MySqlCommand(_checkSlaveCmd, rootConn);
            BinLogInfo info = new BinLogInfo();
            try
            {
                if (!IsMySqlRunning(host, port))
                {
                    return null;
                }
                using (MySqlDataReader reader = cmd.ExecuteReader())
                {
                    if (reader.HasRows)
                    {
                        while (reader.Read())
                        {
                            string relayMasterLogFile = reader["Relay_Master_Log_File"].ToString().ToLower();
                            int execMasterLogPos = Convert.ToInt32(reader["Exec_Master_Log_Pos"]);
                            int pos = relayMasterLogFile.LastIndexOf('.');
                            //there will not be any relay bin log file for a master
                            if (pos == -1)
                            {
                                return null;
                            }
                            string logIdStr = relayMasterLogFile.Substring(pos + 1);
                            int logId = 0;
                            if (!int.TryParse(logIdStr, out logId))
                            {
                                return null;
                            }
                            info.HostInstanceId = instanceId;
                            info.LogId = logId;
                            info.Position = execMasterLogPos;
                            info.EndPoint = ep;
                            return info;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                LogError("Error in GetBinLogInfoForSlave: {0}", ex.Message + ex.StackTrace);
                info = null;
            }
            rootConn.Close();
            return info;
        }

        /// <summary>
        /// Checks if a slave is sync with master
        /// </summary>
        /// <param name="host"></param>
        /// <param name="port"></param>
        /// <returns></returns>
        public bool IsSlaveInSyncWithMaster(string host, int port)
        {
            MySqlConnection rootConn = GetMySqlConnection(host, port);
            MySqlCommand cmd = new MySqlCommand(_checkSlaveCmd, rootConn);
            bool status = false;
            try
            {
                if (!IsMySqlRunning(host, port))
                {
                    return false;
                }
                using (MySqlDataReader reader = cmd.ExecuteReader())
                {
                    if (reader.HasRows)
                    {
                        while (reader.Read())
                        {
                            string masterLogFile = reader["Master_Log_File"].ToString().ToLower();
                            string relayMasterLogFile = reader["Relay_Master_Log_File"].ToString().ToLower();
                            int readMasterLogPos = Convert.ToInt32(reader["Read_Master_Log_Pos"]);
                            int execMasterLogPos = Convert.ToInt32(reader["Exec_Master_Log_Pos"]);

                            if ((masterLogFile.Equals(relayMasterLogFile)) && (readMasterLogPos == execMasterLogPos))
                            {
                                return true;
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                LogError("Error in IsSlaveInSyncWithMaster: {0}", ex.Message + ex.StackTrace);
                status = false;
            }
            rootConn.Close();
            return status;
        }

        /// <summary>
        /// checks if the mysql instance is a master
        /// </summary>
        /// <param name="address"></param>
        /// <param name="port"></param>
        /// <returns></returns>
        public bool CheckIfMaster(IPAddress address, int port)
        {
            try
            {
                MySqlConnection rootConn = GetMySqlConnection(address.ToString(), port);
                LogError("CheckIfMaster: {0}:{1}", address.ToString(), port.ToString());
                if (rootConn == null)
                {
                    LogError("CheckIfMaster: mysql connection null");
                    return false;
                }
                else if (!rootConn.Ping())
                {
                    LogError("CheckIfMaster: Not able to ping mysql connection");
                    return false;
                }

                LogError("Inside CheckIfMaster");

                MySqlCommand cmd = new MySqlCommand(_checkSlaveCmd, rootConn);
                MySqlDataReader reader = null;
                try
                {
                    reader = cmd.ExecuteReader();
                    if (reader.HasRows)
                    {
                        while (reader.Read())
                        {
                            string ioRunning = reader["Slave_IO_Running"].ToString();
                            string sqlRunning = reader["Slave_SQL_Running"].ToString();
                            if (ioRunning == "No" && sqlRunning == "No")
                            {
                                LogError("CheckIfMaster: Is a master");
                                return true;
                            }
                            else
                            {
                                LogError("CheckIfMaster: Not a master");
                                // this is a slave node 
                                return false;
                            }
                        }
                    }
                }
                catch (MySqlException ex)
                {
                    LogError("Error in CheckIfMaster: {0}", ex.Message + ex.StackTrace);
                    return false;
                }
                rootConn.Close();
                LogError("CheckIfMaster: Is a master");
                return true;
            }
            catch (Exception ex)
            {
                LogError("Error in CheckIfMaster: {0}", ex.Message + ex.StackTrace);
                return false;
            }
        }

        /// <summary>
        /// Checks if the mysql instance is running 
        /// </summary>
        /// <param name="endpointName"></param>
        /// <returns></returns>
        public static bool IsMySqlRunning(string address, int port)
        {
            bool isActive = false;
            MySqlConnection rootConn = GetMySqlConnection(address, port);
            if (rootConn == null)
            {
                return false;
            }
            lock (_syncObj)
            {
                try
                {
                    //Pings the mysql server to check if it is active
                    isActive = rootConn.Ping();
                }
                catch (Exception)
                {
                    isActive = false;
                }
            }
            rootConn.Close();
            return isActive;
        }

        /// <summary>
        /// Making the master read-only on shutdown
        /// </summary>
        /// <returns></returns>
        public bool MakeMasterReadOnly()
        {
            bool status = false;
            try
            {
                if (!IsRunning(_endpointName))
                {
                    return false;
                }
                MySqlConnection rootConn = GetConnection(_endpointName);
                //Set read lock on master
                MySqlCommand cmd = new MySqlCommand(_setReadLock, rootConn);
                cmd.ExecuteNonQuery();
                //set the global variable read_only to true
                cmd = new MySqlCommand(_setReadOnly, rootConn);
                cmd.ExecuteNonQuery();
                status = true;
            }
            catch (MySqlException ex)
            {
                LogError("Error in MakeMasterReadOnly: {0}", ex.Message + ex.StackTrace);
                return false;
            }
            return status;
        }

        /// <summary>
        /// log errors
        /// </summary>
        /// <param name="fmt"></param>
        /// <param name="args"></param>
        public void LogError(string fmt, params object[] args)
        {
            _logger.Log(fmt, args);
        }

        /// <summary>
        /// flush log for incremental backup
        /// </summary>
        /// <param name="address"></param>
        /// <param name="port"></param>
        /// <returns></returns>
        public string FlushLog(string address, int port)
        {
            string newLog = string.Empty;
            try
            {
                MySqlConnection rootConn = MySQLClient.GetMySqlConnection(address, port);
                if (rootConn != null && rootConn.Ping())
                {
                    MySqlCommand cmd = new MySqlCommand("FLUSH LOGS", rootConn);
                    cmd.ExecuteNonQuery();
                    cmd = new MySqlCommand("SHOW MASTER STATUS", rootConn);
                    using (MySqlDataReader dataReader = cmd.ExecuteReader())
                    {
                        if (dataReader.HasRows)
                        {
                            while (dataReader.Read())
                            {
                                newLog = dataReader["File"].ToString().ToLower();
                            }
                        }
                    }
                    LogError("FlushLog: SUCCESS: {0}", newLog);
                }
                else
                {
                    LogError("FlushLog: Connection null or not able to ping");
                }
                rootConn.Close();
            }
            catch (MySqlException ex)
            {
                LogError("Error in FlushLog: {0}", ex.Message + ex.StackTrace);
                return newLog;
            }
            return newLog;
        }

        #endregion
    }
}